package com.ys.servers.common.data;

import com.ys.servers.common.util.CommonUtils;
import io.vertx.core.Future;
import io.vertx.core.json.JsonObject;
import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;
import io.vertx.ext.mongo.MongoClient;
import io.vertx.ext.mongo.UpdateOptions;
import io.vertx.redis.RedisClient;
import io.vertx.redis.RedisOptions;
import io.vertx.redis.op.RangeOptions;
import io.vertx.core.Vertx;

import java.util.ArrayList;
import java.util.Calendar;
import java.util.List;

import static com.ys.servers.common.util.CommonUtils.sucAsyncResult;
import static com.ys.servers.common.util.CommonUtils.validAsyncResult;

import io.vertx.redis.op.RangeLimitOptions;

/**
 * Created by taotao on 2017/5/5.
 */
public class DataReader {

    protected static final Logger logger = LoggerFactory.getLogger(DataReader.class.getName());


    private String nameSpace = "YsServer";

    public String getNameSpace() {
        return nameSpace;
    }

    public DataReader setNameSpace(String nameSpace) {
        this.nameSpace = nameSpace;
        return this;
    }

    protected MongoClient mongoClient;
    protected RedisClient redisClient;


    private long expireInterval = 7 * 24 * 3600 * 1000;

    private static DataReader dataReader;

    protected DataReader() {

    }

    public <T extends DataEntity> String getRedisKey(String id, Class<T> clazz) {
        return nameSpace + ":" + clazz.getSimpleName() + ":" + id;
    }

    public <T extends DataEntity> String getMongoCollectionName(Class<T> clazz) {
        return clazz.getSimpleName();
    }

    private void setStoreClient(MongoClient mongoClient, RedisClient redisClient) {
        this.mongoClient = mongoClient;
        this.redisClient = redisClient;
    }


    public static DataReader getInstance() {
        if (dataReader == null) {
            logger.error(" Data Reader has not been initiated. ");
        }
        return dataReader;
    }

    public static void init(Vertx vertx, JsonObject config) {
        if (dataReader != null) {
            return;
        }
        String mongoHost = config.getString("mongo.host", "127.0.0.1");
        int mongoPort = config.getInteger("mongo.port", 27017);
        String nameSpace = config.getString("name.space", "YsServer");
        String redisHost = config.getString("redis.host", "127.0.0.1");
        int redisPort = config.getInteger("redis.port", 6379);
        int redisDb = config.getInteger("redis.db", 0);


        MongoClient mongoClient = MongoClient.createShared(vertx,
                new JsonObject().put("host", mongoHost)
                        .put("port", mongoPort)
                        .put("db_name", nameSpace));

        String name = config.getString("service.name", "");

        RedisClient redisClient = RedisClient.create(vertx,
                new RedisOptions().setHost(redisHost).
                        setPort(redisPort));
        logger.info(name + " DataReader started to init ... \n");


        redisClient.select(redisDb, res -> {
            if (validAsyncResult(res)) {
                StringBuffer initLogInfo = new StringBuffer();
                initLogInfo.append(name + " DataReader  initiated Successfully ! ! \n")
                        .append(" Mongo Host : " + mongoHost + ", port : " + mongoPort + " . \n")
                        .append(" Redis Host : " + redisHost + ", port : " + redisPort + " , Db Index : " + redisDb + " . \n");
                logger.info(initLogInfo);
            } else {
                logger.error(name + " DataReader Failed while initiated ! ! \n");
                logger.error(res.cause());
            }


        });
        dataReader = DataReader.getInstance(mongoClient, redisClient).setNameSpace(nameSpace);

    }

    public static DataReader getInstance(MongoClient mongoClient, RedisClient redisClient) {
        if (dataReader == null) {
            dataReader = new DataReader();
        }
        dataReader.setStoreClient(mongoClient, redisClient);
        return dataReader;
    }

    public void maintainExpire(String key) {
        redisClient.pexpireat(key, Calendar.getInstance().getTimeInMillis() + expireInterval, as -> logger.info("set key " + key + " expire interval"));
    }

    /**
     * @param id
     * @param clazz
     * @param <T>
     * @return
     */
    public <T extends DataEntity> Future<T> get(String id, Class<T> clazz) {
        Future<T> future = Future.future();
        redisClient.get(getRedisKey(id, clazz), ar -> {
            if (ar.succeeded() && ar.result() != null) {
                T dataEntity = DataEntity.parseString(ar.result(), clazz);
                this.maintainExpire(getRedisKey(id, clazz));
                future.complete(dataEntity);
            } else {
                logger.info("redis hit fail");
                mongoClient.findOne(getMongoCollectionName(clazz), new JsonObject().put("_id", id), null, as -> {
                    if (as.succeeded()) {
                        if (as.result() != null) {
                            T dataEntity = DataEntity.parseJsonObject(as.result(), clazz);
                            redisClient.set(getRedisKey(id, clazz), dataEntity.toString(), as1 -> logger.info("insert data to redis"));
                            future.complete(dataEntity);
                        } else {
                            future.complete(null);
                        }

                    } else {
                        logger.error(as.cause());
                        future.fail(as.cause());
                    }
                });
            }
        });
        return future;
    }


    /**
     * 存入db，并且write cache
     *
     * @param dataEntity
     * @param <T>
     * @return
     */
    public <T extends DataEntity> Future<T> save(T dataEntity) {
        Future<T> future = Future.future();
        Class clazz = dataEntity.getClass();
        mongoClient.save(getMongoCollectionName(clazz), dataEntity.toDataObject(), as -> {
            if (as.succeeded()) {
                String id = as.result();
                if (id != null)
                    dataEntity.setId(id);
                redisClient.set(getRedisKey(dataEntity.getId(), clazz), dataEntity.toString(), as1 -> {
                    this.maintainExpire(getRedisKey(dataEntity.getId(), clazz));
                    logger.info("insert data to redis");
                });
            /*
            redisClient.hmset(dataEntity.idKey(dataEntity.getId()),dataEntity.getDataObject(),as1-> {
                this.maintainExpire(dataEntity.idKey(dataEntity.getId()));

                logger.info("insert data to redis");
            });*/
                future.complete(dataEntity);
            } else {
                logger.error(as.cause());
                future.fail(as.cause());
            }


        });
        return future;
    }

    /**
     * 存入数据库 不写缓存
     *
     * @param dataEntity
     * @param <T>
     * @return
     */
    public <T extends DataEntity> Future<T> saveToDB(T dataEntity) {
        Future<T> future = Future.future();
        Class clazz = dataEntity.getClass();
        mongoClient.save(getMongoCollectionName(clazz), dataEntity.toDataObject(), as -> {
            if (as.succeeded()) {
                logger.info("insert data to mongo");
                String id = as.result();
                if (id != null) {
                    dataEntity.setId(id);
                }
                future.complete(dataEntity);
            } else {
                logger.error(as.cause());
                future.fail(as.cause());
            }
        });
        return future;
    }


    /**
     * //判断数据是否在缓存中，如果需要，未命中以后从数据库里读取并建立缓存
     *
     * @param id
     * @param clazz
     * @param findInDb
     * @param <T>
     * @return
     */
    public <T extends DataEntity> Future<Boolean> exist(String id, Class<T> clazz, boolean findInDb) {
        Future<Boolean> future = Future.future();
        try {
            T dataEntity = clazz.newInstance();
            redisClient.exists(getRedisKey(id, clazz), ar -> {
                if (ar.succeeded() && ar.result() != null) {
                    future.complete(true);
                } else {
                    logger.info("redis hit fail");
                    //如果要求尝试读取db，则读取db并尝试建立缓存
                    if (findInDb) {
                        getFromDB(id, clazz, true).setHandler(res -> {
                            if (res.succeeded()) {
                                future.complete(true);
                            } else {
                                future.fail(res.cause());
                            }
                        });
                    }
                }
            });
        } catch (InstantiationException e) {
            e.printStackTrace();
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        }
        return future;
    }

    /**
     * 初始化缓存时候，首先从db读取，如果需要，读取后建立缓存。
     *
     * @param id
     * @param clazz
     * @param cached
     * @param <T>
     * @return
     */
    public <T extends DataEntity> Future<T> getFromDB(String id, Class<T> clazz, boolean cached) {
        Future<T> future = Future.future();
        mongoClient.findOne(getMongoCollectionName(clazz), new JsonObject().put("_id", id), null, as -> {
            if (as.succeeded()) {
                T dataEntity = DataEntity.parseJsonObject(as.result(), clazz);
                if (cached)
                    //暂时不处理插入redis失败，只要数据库读取成功，就认为读取成功。
                    saveToCache(dataEntity);
                future.complete(dataEntity);
            } else {
                logger.error(as.cause());
                future.fail(as.cause());
            }
        });
        return future;

    }

    /**
     * 只存入缓存
     *
     * @param dataEntity
     * @param <T>
     * @return
     */
    public <T extends DataEntity> Future<Boolean> saveToCache(T dataEntity) {
        Future<Boolean> future = Future.future();
        Class clazz = dataEntity.getClass();
        String storeKey = getRedisKey(dataEntity.getId(), clazz);
        redisClient.set(storeKey, dataEntity.toString(), as -> {
            if (as.succeeded()) {
                this.maintainExpire(storeKey);
                logger.info("Success inserting data to redis with id : " + storeKey);
                future.complete(true);
            } else {
                logger.info("Fail to insert data to redis with id : " + storeKey);
                future.fail(as.cause());
            }
        });
        return future;
    }


    public <T extends DataEntity> Future<Boolean> remove(String id, Class<T> clazz) {
        Future<Boolean> future = Future.future();
        mongoClient.removeDocument(getMongoCollectionName(clazz), new JsonObject().put("_id", id), as -> {
            if (as.succeeded()) {
                redisClient.del(getRedisKey(id, clazz), redisAs -> {
                    if (redisAs.succeeded()) {
                    } else {
                        //future.fail(res.cause());
                    }
                    future.complete();
                });
            } else {
                future.fail(as.cause());
            }
        });
        return future;
    }

    /**
     * 失效一条记录，该记录从存储列表删除，更换一个id存档在数据库里,并记录存档的时间戳。
     *
     * @param id
     * @param clazz
     * @param <T>
     * @return
     */
    public <T extends DataEntity> Future<Boolean> deprecate(String id, Class<T> clazz) {
        Future<Boolean> future = Future.future();
        get(id, clazz).setHandler(getResult -> {
            if (validAsyncResult(getResult)) {
                T dataEntity = getResult.result();
                mongoClient.save(getMongoCollectionName(clazz) + ":Deprecated",
                        dataEntity.toDataObject().put("deprecateTime", System.currentTimeMillis()), as -> {
                            if (sucAsyncResult(as)) {
                                remove(id, clazz).setHandler(removeAs -> {
                                    if (sucAsyncResult(removeAs)) {
                                        future.complete(true);
                                    } else {
                                        future.fail(removeAs.cause());
                                    }
                                });
                            } else {
                                future.fail(as.cause());
                            }
                        });
            } else {
                future.fail(getResult.cause());
            }
        });
        return future;
    }


    public <T extends DataEntity> Future<T> restore(String id, Class<T> clazz) {

        Future<T> future = Future.future();
        String deprecatedMongoName = getMongoCollectionName(clazz) + ":Deprecated";
        mongoClient.findOne(deprecatedMongoName, new JsonObject().put("_id", id), null, as -> {
            if (CommonUtils.validAsyncResult(as)) {
                T dataEntity = DataEntity.parseJsonObject(as.result(), clazz);
                dataEntity.setId(id);
                mongoClient.removeDocument(deprecatedMongoName, new JsonObject().put("_id", id), remRes -> {
                    if (CommonUtils.sucAsyncResult(remRes)) {
                        this.save(dataEntity);
                        future.complete(dataEntity);
                    } else {
                        future.fail(remRes.cause());
                    }
                });

            } else {
                future.fail(as.cause());
            }
        });
        return future;

    }

    /**
     * 递增数据库里的某个字段
     *
     * @param collectionName
     * @param id
     * @param field
     * @param deltaValue
     * @param <T>
     * @return
     */
    public <T extends DataEntity> Future<Boolean> increaseFieldInDb(String collectionName, String id, String field, int deltaValue) {
        Future<Boolean> future = Future.future();
        JsonObject query = new JsonObject().put("_id", id);
        JsonObject increase = new JsonObject().put("$inc", new JsonObject().put(field, deltaValue));
        mongoClient.updateCollection(collectionName, query, increase, res -> {
            if (res.succeeded()) {
                future.complete();
            } else {
                future.fail(res.cause());
            }
        });
        return future;
    }

    /**
     * Added by xingyt
     * 递增一个字段，并刷新缓存
     *
     * @param id
     * @param field
     * @param deltaValue
     * @param clazz
     * @param <T>
     * @return
     */
    public <T extends DataEntity> Future<Boolean> increaseFieldInDb(String id, String field, int deltaValue, Class<T> clazz) {
        Future<Boolean> future = Future.future();
        try {
            DataEntity dataEntity = clazz.newInstance();
            JsonObject query = new JsonObject().put("_id", id);
            JsonObject increase = new JsonObject().put("$inc", new JsonObject().put(field, deltaValue));
            mongoClient.updateCollection(dataReader.getMongoCollectionName(clazz), query, increase, res -> {
                if (res.succeeded()) {
                    String storeKey = dataReader.getRedisKey(id, clazz);
                    //先让缓存失效，后面在优化
                    redisClient.del(storeKey, res2 -> {
                        logger.info("");
                    });
                    future.complete();
                } else {
                    future.fail(res.cause());
                }
            });

        } catch (InstantiationException e) {
            e.printStackTrace();
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        }
        return future;
    }

    /**
     * @param name
     * @param id
     * @param deltaValue
     * @return
     */
    public Future<Boolean> increaseSimpleData(String name, String id, long deltaValue) {
        Future<Boolean> future = Future.future();
        JsonObject query = new JsonObject().put("_id", id);
        JsonObject increase = new JsonObject().put("$inc", new JsonObject().put("data", deltaValue));
        mongoClient.updateCollectionWithOptions(name, query, increase, new UpdateOptions(true), res -> {
            if (res.succeeded()) {
                String redisKey = nameSpace + ":" + name + ":" + id;
                //先让缓存失效，后面在优化
                redisClient.incrby(redisKey, deltaValue, res2 -> {
                    future.complete(true);
                });
            } else {
                future.fail(res.cause());
            }
        });

        return future;
    }

    /**
     * 同时在数据库里递增更新多个字段，用于私有数据更新，此类数据只有个人会写，更新后会使缓存失效。
     *
     * @param id
     * @param fields
     * @param deltaValues
     * @param clazz
     * @param <T>
     * @return
     */
    public <T extends DataEntity> Future<Boolean> increaseMultiFieldInDb(String id, List<String> fields, List<Integer> deltaValues, Class<T> clazz) {
        Future<Boolean> future = Future.future();
        try {
            DataEntity dataEntity = clazz.newInstance();
            JsonObject query = new JsonObject().put("_id", id);

            //JsonObject increase = new JsonObject();//.put("$inc", new JsonObject().put(field, deltaValue));
            JsonObject increaseFieldsAndValues = new JsonObject();
            int l = fields.size();
            for (int i = 0; i < l; i++) {
                increaseFieldsAndValues.put(fields.get(i), deltaValues.get(i));
            }
            JsonObject increase = new JsonObject().put("$inc", increaseFieldsAndValues);
            mongoClient.updateCollection(getMongoCollectionName(clazz), query, increase, res -> {
                if (res.succeeded()) {
                    String storeKey = getRedisKey(id, clazz);
                    redisClient.del(storeKey, res2 -> {
                    });
                    future.complete();
                } else {
                    future.fail(res.cause());
                }
            });

        } catch (InstantiationException e) {
            e.printStackTrace();
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        }
        return future;
    }


    /**
     * 。。。还没写完。
     *
     * @param id
     * @param field
     * @param deltaValue
     * @param clazz
     * @param <T>
     * @return
     */
    @Deprecated
    public <T extends DataEntity> Future<T> increaseFieldInDbAndFetch(String id, String field, int deltaValue, Class<T> clazz) {
        Future<T> future = Future.future();
        JsonObject increase = new JsonObject().put("$inc", new JsonObject().put(field, deltaValue));
        String col = getMongoCollectionName(clazz);
        JsonObject query = new JsonObject().put("_id", id);
        String queryString = "{\"_id\":\"" + id + "\"}";
        String updateString = "{\"$inc\":{\"" + field + "\":" + deltaValue + "}}";
        System.out.println(queryString);
        System.out.println(updateString);
        JsonObject command = new JsonObject().put("findAndModify", col).put("query", queryString).put("update", updateString);
        System.out.println(command);
        mongoClient.runCommand("findAndModify", command, res -> {
            if (res.succeeded()) {
                T dataEntity = DataEntity.parseJsonObject(res.result(), clazz);
                future.complete(dataEntity);
            } else {
                future.fail(res.cause());
            }
        });

        return future;
    }

    /**
     * Added by xingyt.
     *
     * @param id
     * @param field
     * @param score
     * @param maxNumber
     * @param <T>
     * @return
     */
    public <T extends DataEntity> Future<Boolean> addSortedList(String id, String field, long score, long maxNumber) {
        Future<Boolean> future = Future.future();
        /*
        实现A：这个实现简洁，但是存在末尾并列时候，后来者按照字典序有可能顶替掉先达到者。
        redisClient.zadd(id, score, field, res -> {
            redisClient.zcard(id, countRes -> {
                if (validAsyncResult(countRes)) {
                    if (countRes.result() > maxNumber) {
                        redisClient.zremrangebyrank(id, 0, 0, remRes -> {
                        });
                    }
                }
            });
        });*/
        redisClient.zcard(nameSpace + ":" + id, countRes -> {
            if (validAsyncResult(countRes)) {
                if (countRes.result() < maxNumber)
                    //直接插
                    doRealAddSortedListAndCompleteFuture(id, field, score, countRes.result(), maxNumber, future);
                else {
                    //首先取得最后一名的scrore，如果相当，先到者保持，后来者不能加入
                    redisClient.zrangeWithOptions(nameSpace + ":" + id, 0, 0, RangeOptions.WITHSCORES, oneRes -> {
                        if (validAsyncResult(oneRes)) {
                            int minScore = Integer.parseInt(oneRes.result().getList().get(1).toString());
                            if (minScore < score)
                                doRealAddSortedListAndCompleteFuture(id, field, score, countRes.result(), maxNumber, future);
                        } else {
                            future.fail(oneRes.cause());
                        }
                    });
                }
            } else {
                future.fail(countRes.cause());
            }

        });

        return future;
    }

    private void doRealAddSortedListAndCompleteFuture(String id, String field, long score, long currentCount, long maxNumber, Future<Boolean> future) {
        redisClient.zadd(nameSpace + ":" + id, score, field, addRes -> {
            if (addRes.succeeded()) {
                if (currentCount >= maxNumber) {
                    redisClient.zremrangebyrank(id, 0, 0, remRes -> {
                    });
                }
                future.complete(true);
            }

        });
    }

    /**
     * 在redis里获取一个有序排名，自己设置要不要获取分数。
     *
     * @param id
     * @param start
     * @param stop
     * @param options
     * @return
     */
    public Future<List<String>> getAllInSortedList(String id, long start, long stop, RangeOptions options) {
        Future<List<String>> future = Future.future();
        redisClient.zrangeWithOptions(nameSpace + ":" + id, start, stop, options, res -> {
            List<String> sortedList = new ArrayList<String>();
            if (res.succeeded() && res.result() != null) {
                res.result().getList().forEach(value -> {
                    sortedList.add((String) value);
                });
            }
            future.complete(sortedList);
        });
        return future;
    }

    /**
     * 在Redis里获取一个有序排名。含有分数。
     *
     * @param id
     * @param start
     * @param stop
     * @return
     */
    /*public Future<List<SortedRedisListItem>> getAllInSortedListWithScores(String id, long start, long stop) {
        Future<List<SortedRedisListItem>> future = Future.future();
        getAllInSortedList(id, start, stop, RangeOptions.WITHSCORES).setHandler(res -> {
            List<SortedRedisListItem> sortedRedisListItem = new ArrayList<SortedRedisListItem>();
            if (res.succeeded() && res.result() != null)
                sortedRedisListItem = SortedRedisListItem.genReverseSortedListWithScoreFromRedisSearch(res.result());
            future.complete(sortedRedisListItem);

        });
        return future;
    }*/

    /**
     * 在Redis里获取一个有序排名对象。含有分数。
     *
     * @param id
     * @param start
     * @param stop
     * @return
     */
    public <T extends DataEntity> Future<List<T>> getAllDataInSortedListWithScores(String id, long start, long stop, Class<T> clazz) {
        Future<List<T>> future = Future.future();
        String key = getRedisKey(id, clazz);
        redisClient.zrange(key, start, stop, res -> {
            List<T> sortedRedisListItem = new ArrayList<>();
            if (validAsyncResult(res)) {
                res.result().forEach(value -> {
                    try {
                        T dataEntity = DataEntity.parseString(value.toString(), clazz);
                        if (dataEntity != null) {
                            sortedRedisListItem.add(dataEntity);
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                });
            }
            future.complete(sortedRedisListItem);
        });
        return future;
    }

    /**
     * 在Redis里获取一个有序排名对象。含有分数。
     *
     * @param id
     * @param start
     * @param stop
     * @return
     */
    public <T extends DataEntity> Future<List<T>> getAllDataInSortedListByScores(String id, long start, long stop, Class<T> clazz) {
        Future<List<T>> future = Future.future();
        String key = getRedisKey(id, clazz);
        redisClient.zrangebyscore(key, Long.toString(start), Long.toString(stop), RangeLimitOptions.NONE, res -> {
            List<T> sortedRedisListItem = new ArrayList<>();
            if (validAsyncResult(res)) {
                res.result().forEach(value -> {
                    try {
                        T dataEntity = DataEntity.parseString(value.toString(), clazz);
                        if (dataEntity != null) {
                            sortedRedisListItem.add(dataEntity);
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                });
            }
            future.complete(sortedRedisListItem);
        });
        return future;
    }

    /**
     * Added by xingyt 从数据库里获取按照某个字段排名的名次
     *
     * @param queryField
     * @param queryValue
     * @param comparefield
     * @param compareValue
     * @param clazz
     * @param <T>
     * @return
     */
    public <T extends DataEntity> Future<Long> getRankInDb(String queryField, String queryValue,
                                                           String comparefield, long compareValue, Class<T> clazz) {
        Future<Long> rank = Future.future();
        mongoClient.count(getMongoCollectionName(clazz), new JsonObject().put(queryField, queryValue).
                put(comparefield, new JsonObject().put("$gt", compareValue)), res -> {
            long result = -1;
            if (validAsyncResult(res)) {
                result = res.result();
            }
            rank.complete(result + 1);
        });

        return rank;
    }

    public Future<Boolean> incrByInRedis(String key, float value) {
        Future<Boolean> future = Future.future();
        redisClient.incrbyfloat(nameSpace + ":" + key, value, incrRes -> {
            if (incrRes.succeeded())
                future.complete(true);
            else
                future.complete(false);
        });
        return future;
    }

    public Future<Boolean> simpleSetInRedis(String key, String value) {
        Future<Boolean> future = Future.future();
        redisClient.set(nameSpace + ":" + key, value, incrRes -> {
            if (incrRes.succeeded())
                future.complete(true);
            else
                future.complete(false);
        });
        return future;
    }

    public Future<String> simpleGetFromRedis(String key) {
        Future<String> future = Future.future();
        redisClient.get(nameSpace + ":" + key, res -> {
            if (res.succeeded())
                future.complete(res.result());
            else
                future.fail(res.cause());
        });
        return future;
    }

    public Future<Boolean> simpleRemoveFromRedis(String key) {
        Future<Boolean> future = Future.future();
        redisClient.del(nameSpace + ":" + key, res -> {
            if (res.succeeded())
                future.complete(true);
            else
                future.fail(res.cause());
        });
        return future;
    }

    public <T extends DataEntity> Future<Boolean> addToSortedListAndDB(String redisListId, T dataEntity) {
        Future<Boolean> future = Future.future();
        Class clazz = dataEntity.getClass();
        mongoClient.save(getMongoCollectionName(clazz), dataEntity.toDataObject(), as -> {
            if (as.succeeded()) {
                String id = as.result();
                if (id != null) {
                    dataEntity.setId(id);
                }
                addToRedisSortedList(redisListId, dataEntity).setHandler(res -> {
                    if (CommonUtils.validAsyncResult(res)) {
                        future.complete(true);
                    } else {
                        logger.error(res.cause());
                        future.fail(res.cause());
                    }
                });
            } else {
                logger.error(as.cause());
                future.fail(as.cause());
            }

        });
        return future;
    }

    public <T extends DataEntity> Future<Boolean> addToRedisSortedList(String redisListId, T dataEntity) {
        Future<Boolean> future = Future.future();
        redisClient.zadd(getRedisKey(redisListId, dataEntity.getClass()), dataEntity.getScore(), dataEntity.toString(), addRes -> {
            if (addRes.succeeded()) {
                future.complete(true);
            } else {
                logger.error(addRes.cause());
                future.fail(addRes.cause());
            }
        });


        return future;
    }

    public <T extends DataEntity> Future<Boolean> removeFromRedisSortedList(String redisListId, T dataEntity) {
        Future<Boolean> future = Future.future();
        redisClient.zrem(getRedisKey(redisListId, dataEntity.getClass()), dataEntity.toString(), remRes -> {
            if (remRes.succeeded()) {
                future.complete(true);
            } else {
                logger.error(remRes.cause());
                future.fail(remRes.cause());
            }
        });


        return future;
    }

    public <T extends DataEntity> Future<Long> removeFromRedisSortedListByScore(String redisListId, Class<T> clazz, double min, double max) {
        Future<Long> future = Future.future();
        redisClient.zremrangebyscore(getRedisKey(redisListId, clazz), Double.toString(min), Double.toString(max), remRes -> {
            if (remRes.succeeded()) {
                future.complete(remRes.result());
            } else {
                logger.error(remRes.cause());
                future.fail(remRes.cause());
            }
        });


        return future;
    }

}
